package com.spx.chapter05.sink;

import com.spx.chapter05.pojo.Event;
import com.spx.util.SampleDataUtil;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * create by undeRdoG on  2022-05-01  16:09
 * 凡心所向，素履以往，生如逆旅，一苇以航。
 */
public class MySQLSinkTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStreamSource<Event> dataSource = env.fromCollection(SampleDataUtil.getSample());

        dataSource.addSink(JdbcSink.sink(
                "insert into `flink_sink`.`event` values (?,?,?)",
                (statement,event) -> {
                 statement.setString(1, event.user);
                 statement.setString(2, event.url);
                 statement.setLong(3, event.timestamp);
                },
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withDriverName("com.mysql.jdbc.Driver")
                .withUrl("jdbc:mysql://localhost:3306/flink_sink")
                .withPassword("root")
                .withUsername("root").build()
                ));


        env.execute();

    }
}
